package net.bwie.realtime.dws.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.dws.log.bean.AverageSpeed;
import net.bwie.realtime.dws.log.bean.MonitorInfo;
import net.bwie.realtime.dws.log.function.SpeedWindowFunction;
import net.bwie.realtime.dws.log.utils.DorisUtil1;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import utils.KafkaUtil;

public class AverageSpeedMonitor {

    public static void main(String[] args) throws Exception{

        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);

        //2.读取数据源
        DataStream<String> carStream = KafkaUtil.consumerKafka(env, "dwd-traffic-monitor-log");
//      carStream.print("kafka");
        //3.数据处理
        DataStream<String> resultStream = handle(carStream);
        resultStream.print("result");

        //4.数据输出
        DorisUtil1.saveToDoris(resultStream,"vehicle_realtime_report","dws_average_speed_report");

        //5.执行
        env.execute("AverageSpeedMonitor");
    }

    private static DataStream<String> handle(DataStream<String> stream) {

        SingleOutputStreamOperator<MonitorInfo> mapStream = stream.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String value) throws Exception {
                return JSON.parseObject(value, MonitorInfo.class);
            }
        });



        // 窗口 滑动窗口，窗口长度是为5分钟，滑动步长为1分钟
        WindowedStream<MonitorInfo, String, TimeWindow> windowStream = mapStream.keyBy(v -> v.getMonitorId())
                .window(SlidingProcessingTimeWindows.of(Time.minutes(5), Time.minutes(1)));

        //聚合计算
        SingleOutputStreamOperator<AverageSpeed> applyStream = windowStream.apply(new SpeedWindowFunction());


        SingleOutputStreamOperator<String> resultStream = applyStream.map(averageSpeed -> {
            return String.format(
                    "{\"window_start_time\":\"%s\",\"window_end_time\":\"%s\",\"cur_date\":\"%s\"," +
                            "\"monitor_id\":\"%s\",\"avg_speed\":\"%s\",\"car_count\":\"%s\"}",
                    averageSpeed.getStartTime() != null ? averageSpeed.getStartTime() : "",
                    averageSpeed.getEndTime() != null ? averageSpeed.getEndTime() : "",
                    averageSpeed.getStartTime() != null ? averageSpeed.getStartTime().substring(0, 10) : "",
                    averageSpeed.getMonitorId() != null ? averageSpeed.getMonitorId() : "",
                    averageSpeed.getAvgSpeed() != 0 ? averageSpeed.getAvgSpeed() : "",
                    averageSpeed.getCarCount() != 0 ? averageSpeed.getCarCount() : ""
            );
        });
        return resultStream;
    }
}
